package gbench.appdemo.myworld.junit.eco;

import gbench.appdemo.myworld.economy.Economy;
import gbench.appdemo.myworld.util.Latch;
import gbench.common.tree.LittleTree.IRecord;
import gbench.common.tree.LittleTree.SQL;
import gbench.commonApp.data.DataApp;

import static gbench.common.tree.LittleTree.IRecord.*;
import static gbench.common.tree.LittleTree.SQL.nsql;
import static gbench.common.tree.LittleTree.*;
import static gbench.common.tree.LittleTree.KVPair.*;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.junit.jupiter.api.Test;


/**
 * 构造一个经济体:与 Junit4的却别是，es = Executors.newCachedThreadPool() 改用Cached线程池，以开启最大并行度<br>
 * 防止 当线程数达到上限（5）之后，其余的数据都 被积压到一个线程（95万的数量）中进行处理的情况。
 * @author gbench
 *
 */
public class JunitE5 {
    
    /**
     * 计时器
     * @param runnable 待运行的程序
     * @return 历时时间
     */
    public static synchronized double elapse(final Runnable runnable) {
        return elapse(runnable,false);
    }

    /**
     * 计时器
     * @param runnable 待运行的程序
     * @param b 是否打印统计信息
     * @return 持续时间
     */
    public static synchronized double elapse(final Runnable runnable,final boolean b) {
        final long begTime = System.nanoTime();
        runnable.run();
        final var last = (System.nanoTime()-begTime)/Math.pow(10, 6);
        if(b) System.err.println(MFT("lasts for:{0} ms",last));
        return last;
    }

    /**
     * 发送消息
     * @param eco 经济体
     */
    public void send(final Economy eco) {
        while(true) {
            final var flag = ai.get()>=MAX_SIZE; // 结束标记
            final var proto = REC("stopflag",flag);
            eco.publish(proto.derive("producer",1,"id",ai.getAndIncrement(),
                "name",rnd(names),"drcr",Math.random()>0.5?1:0,"amount",rand.nextInt(10000)
            ));
            if(flag) break; // 停止运行信号
        }// while
    }
    
    /**
     * 保存 event 消息事件
     * @param event 消息事件
     */
    public void persist(final IRecord event) {
        
        // 需要注意: 一般sqls.add(...) 不要写在es.execute，否则 就会 出现数据(event) 被遗漏的可能:
        // 1) 一种可性是 A线程  尚未把 SQL复制到 batchsqls时, B线程给予切入，B线程 接收到 停止信(stopflag)，shutdown 掉 economy,
        //    导致  program_latch 栓锁 被拉开，主线程 终止 掉程序。 进而 导致 A线程 由于 程序已被终止，被迫 由OS 强行提前关闭，造成 数据插入不完全而出现event数据遗失。
        // 2) 还有 一种可能就是  es 由于 线程数达 到上限 无法 开启 新的线程来接收 event, 造成无法创建接新计算任务来接收event, 结果 就是 遗漏掉了 event数据
        synchronized(sqls) { // 在 SQL上进行同步
            final var data = A( event.get("id"), event.get("drcr"), event.get("name"), event.get("amount"), LocalDateTime.now()); // 数据参数
            sqls.add(nsql(inserttpl(null),REC("entries",L(rb.get(data)))));// 加入SQL的缓存,这是一个采用莫仍表名的插入语句。
        }// synchronized
       
        es.execute(()->{ // 刷新数据
            task_runnings.incrementAndGet();// 标记一个正在运行的任务
            final var threadLocal = new ThreadLocal<List<SQL>>();// 线程本地存储
            synchronized(sqls) { 
                if( sqls.size()>=BATCH_SIZE || event.bool("stopflag")) { // 分批插入 的时机 获取
                    final var aa = new ArrayList<>(sqls); // 提取累积sql 语句
                    threadLocal.set(aa);// 把SQL复制到本地线程存储以准备数据吃就好
                    sqls.clear(); // 清空 SQL列表缓存
                }//if 批次写入
            }// sqls
            
            if( threadLocal.get()!=null && threadLocal.get().size()>0 ) {
                final var batchsqls= threadLocal.get();
                final var entriesMap = batchsqls.stream()
                    .flatMap(e->e.getSqlCtxAsOneRecord().lla("entries", IRecord.class).stream())
                    .collect(Collectors.groupingBy(e->e.str("name")));
                final var names = entriesMap.keySet().toArray(String[]::new);// 提取产品名称 数组
                
                // 打印线程信息的任务摘要
                System.out.println(MFT("+thread name:{0},pct name:{1}, batchsqls size:{2}",
                    Thread.currentThread().getName(),Arrays.asList(names),batchsqls.size())) ;
                
                //
                db_tbls(names).map( e -> REC ( // 根据产品名称 给与分配 数据 访问资源。
                    "name",e.str("name"), // 产品名称
                    "tblname", e.str("tblname"), // 表名
                    "entries", entriesMap.get(e.str("name")), // 插入的数据内容
                    "dataApp", new MyDataApp(e.str("dbname")) // 数据访问对象
                )).forEach( e -> { // 批量的插入数据。
                    
                    final var name = e.str("name");// 产品名称
                    final var tblname = e.str("tblname"); // 表名
                    final var entries = e.get("entries"); // 插入的数据内容
                    final var dataApp = e.get("dataApp", MyDataApp.class); // 数据访问对象
                    
                    System.err.println(MFT("db:{0},table:{1},name:{2},size:{3},lasts:{4} ms",
                        dataApp.getName(),tblname,name,((List<?>)entries).size(),
                        elapse( ()->{ // 他统计计时
                            dataApp.withTransaction( sess -> { // 执行数据库插入操作
                                sess.sqlexecute("set unique_checks=0"); // 关闭唯一性检查
                                Stream.of(nsql(inserttpl(tblname),REC("entries",entries))).forEach(trycatch2(sess::sqlexecute)); // 插入数据
                                sess.sqlexecute("set unique_checks=1"); // 开启唯一性检查
                            }); // withTransaction
                        }) // timeit
                    )); // System
                }); // withTransaction
            }// if
            
            if(event.bool("stopflag")) {// 接收到停止标记
                while( task_runnings.get()!=1 ) { // 等待其他任务运行结束
                    System.out.println(MFT("尚有:{0}个其余线程正在运行,等待他们运行结束！",task_runnings.get()-1));
                    Stream.of(1000).forEach(trycatch3(Thread::sleep));// 等待其他的任务执行完毕。
                }// 等待
                latch.countdown(); // 发送停止运行 命令
            }else {
                task_runnings.decrementAndGet(); // 正在运行的tasks数减1
                // System.out.println(MFT("-thread name:{0},pct name:{1}, 运行结束！",Thread.currentThread().getName(),Arrays.asList(names))) ;
            }//if
        }); //  es.execute  
    }
    
    /**
     * 消息接收函数
     * @param event
     */
    public void receive(final IRecord event) {
        //System.out.println("1)"+event);
        storage.add(event);// 添加入storage
        this.persist(event);
    }
    
    /**
     * 数据库与数据表名
     * @param keys 产品键名列表
     * @return [{dbname,tblname}]
     */
    public Stream<IRecord> db_tbls(final String keys[]){
        final var kk = keys == null?names2dbs.keys().toArray(String[]::new):keys;
        return names2dbs.filter(kk).stream(kvp->kvp.map2((IRecord rec)->rec.filter("dbname,tblname,name")));
    }
    
    /**
     * 
     */
    @Test
    public void foo() {
        
        System.out.println(MFT("分库分表的数据库演示(运行时间2分钟左右 ,数据库需要事先创建好而数据表不需要)\n"
            + "基本运行参数:\tMax Size:[{0}]\tBatch Size:[{1}]\n\n"
            + "分配策略如下:",this.MAX_SIZE,this.BATCH_SIZE));
        
        db_tbls(null).map(e->REC(
            "tblname", e.str("tblname"),
            "dataApp", new MyDataApp(e.str("dbname")),
            "name",e.str("name")
        )).forEach( e -> {
            final var dataApp = e.get("dataApp",MyDataApp.class);
            final var tblname = e.str("tblname");
            final var name = e.str("name");
            final SQL.Builder sb = SQL.builder(tblname); // 狗仔entry sql 系列构建器具
            final List<String> createsqls = sb.get(proto).createTable(true); // 生成createsqls
            dataApp.withTransaction(sess->TUPLES(createsqls).filter(t->t._1()!=1).map(t->t._2()).forEach(trycatch2(sess::sqlexecute)));// 创建数据表
            System.out.println(MFT("{0}/{1}/{2}",name,tblname,dataApp.getName()));
        });
        
        System.out.println("\n开始数据插入实验:");
        new Economy(1024) // 构造一个经济体
        .produce(this::send) // 发送消息
        .consume(this::receive) // 接收消息
        .collect(eco -> {
            latch.await();// 等待producer 运行结束
            eco.shutdown();
            es.shutdown();
            System.out.println("经济体 停止运行");
            program_latch.countdown();
        });
        
        program_latch.await();
        
        // 查看数据结果
        TUPLES( 
            storage.stream(), // 基础库存
            db_tbls(null).flatMap(e->new MyDataApp(e.str("dbname")).sql2stream(MFT("select * from {0}",e.str("tblname"))))  
        ).forEach(JunitE5::stats); // 对name 进行透视
        
        System.out.println("程序运行结束");
    }
    
    @Test
    public void bar() {
        // 查看数据结果
        TUPLES( 
            storage.stream(), // 基础库存
            db_tbls(null).flatMap(e->new MyDataApp(e.str("dbname")).sql2stream(MFT("select * from {0}",e.str("tblname"))))  
        ).forEach(JunitE5::stats); // 对name 进行透视
    }
    
    /**
     * 数据统计
     * @param tup 数据分组(分组编号,数据流项目)
     */
    public static void stats(final Tuple2<Long,Stream<IRecord>> tup) {
        System.out.println("-------"+tup._1()+"-------");
        elapse(()->{//  统计计时
            tup._2().collect(pvtclc("name,drcr")).dfs_eval_forall2((path,rr)-> KVP ( 
                Arrays.asList(("E5"+path).split("/")), // 构造路径序列
                rr.collect(Collectors.summarizingDouble(e->((IRecord)e).dbl("amount"))) // 路净值
            )).collect( ssclc( compose_f( IRecord::MAP2REC, IRecord::KVPSMERGE2 ) ) ) // 提取路径层级并 展缓恒树形节点
            .treeNode().forEach( node -> { // 转换成树形结构节点。
                final var dss = node.attr("value",DoubleSummaryStatistics.class); // 提取统计结果
                System.out.println(" -- ".repeat(node.getLevel()-1)+ node.getPath()+"\t"+ (dss==null?"-":dss.getSum()+"/"+dss.getCount()) );
            });// treeNode
        },true);// timeit
    }
    
    /**
     * 数据操作类
     * @author gbench
     *
     */
    final class MyDataApp extends DataApp {
        
        public MyDataApp(final String dbname) { 
            this.dbname = dbname;
            this.reload(this.configureDbName(dbname));
        }
        
        public IRecord configureDbName(final String dbname) { return 
            cfg.derive("url", MFT("jdbc:mysql://localhost:3306/{0}?serverTimezone=GMT%2B8", dbname));
        }
        
        public List<IRecord> sql2records(String sql){
            final var ll = new LinkedList<IRecord>();
            this.withTransaction(sess->ll.addAll(sess.sql2records(sql)));
            return ll;
        }
        
        public Stream<IRecord> sql2stream(String sql){
           return this.sql2records(sql).stream();
        }
        
        public String getName() {
            return this.dbname;
        }
        
        private String dbname;
    }; // MyDataApp 数据数据库操作类
    
    /**
     * 根据表名提取 数据模版
     * @param table 表名
     * @return 数据插入模版
     */
    public String inserttpl(final String table) {
        final String inserttpl = "insert into " + (table==null?"t_tbl":table) + " (" + s1 + ") values ${ foreach e in entries (" + s2 + ") }";
        return inserttpl;
    }
    
    private final Random rand = new Random(); // 随机器生成器
    private final int BATCH_SIZE = 10000; // 插入数据的批次大小
    private final int MAX_SIZE = 1000000; // 插入数据总量
    private final List<SQL> sqls = new LinkedList<>(); // SQL 列表
    private final List<IRecord> storage = new LinkedList<>(); // 消息缓冲,历史的数据记录
    
    private final String[] names = "甲苯,甲醇,乙醇,丁二醇".split(","); // 产品名称
    private final IRecord proto = REC("#id",1,"~code",1,"drcr",1,"name","甲苯","amount",2d,"createtime",LocalDateTime.now());// 原型结构
    private final List<String> keys = proto.aoks2rec(compose_f(e->e.str("name"),SQL::parseFieldName)).filter2("id").keys(); // 提取键名列表
    private final IRecord.Builder rb = IRecord.builder(proto.filter2("#id")); // 去除id
    
    private final String TBLNAME_PREFIX = "T_DOUBLE_ENTRY";// 表名前缀
    private final String s1 = JOIN( keys, "," ); // 设置键名
    private final String s2 = JOIN( keys.stream().map(e -> "e." + e), "," ); // 键名
    private final IRecord names2dbs = REC(GROUP2MAP(Stream.of(names).map(kvp()).map( kvp -> REC ( // 数据表配置,数据存储的分配策略
        "id", kvp.key(), // id 编号
        "name",kvp.value(), // 产品名称
        "tblname", MFT("{0}_{1}", TBLNAME_PREFIX,kvp.key()), // 表名
        "dbname", MFT("mydb{0}", kvp.key() % 2 + 1) // 数据库名
    )), e -> e.str("name"))); // 采用名称生成 数据表
    
    private final AtomicInteger ai = new AtomicInteger(0); //计数器
    private final Latch program_latch = new Latch(1); // 程序运行 的 1道栓所
    private final Latch latch = new Latch(1);// 生产活动控制锁, latch 需要用生产者的数量为基础
    private final ExecutorService es = Executors.newCachedThreadPool();// 线程池,不限定工作者者数量
    private final AtomicInteger task_runnings = new AtomicInteger(0); // 正在运行的task的数量
}
